import java.util
import org.apache.spark.mllib.recommendation.{ALS, Rating}
import org.apache.spark.{SparkConf, SparkContext}
 
object MyALS {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("MYALS")
    val sc = new SparkContext(conf)
    /**
      * 数据数据：来自【用户产品统计数据】：{产品ID,用户ID,PV(访问量),产品类型}
      * car000466  |513049888252|1|car
        house000028|513041784009|1|house
        house000044|4510174246|2|house
        house000044|513042437238|1|house
        house000144|513040816140|11|house
        house000144|523047653102|1|house
        house000365|513041784009|1|house
      */
    val protj_rdd = sc.textFile("file:///D://测试数据//dx/07.DxProStatistic/part-r-00*")
 
    /* 处理数据：将用户产品统计中产品id-->整型产品ID(HashCode)和用户ID-->整型用户ID(HashCode)
	   * 数据样例为三元组：用户id|用户IDHash码,产品id|产品IDHash码,PV(访问量)|产品类型
	   * (5064062000|2042345,car000044|000044,5|car)
	   */
    val basicId = protj_rdd.map(x => {
      val thrarray = x.split("\\|")
      (thrarray(1)+"|"+thrarray(1).hashCode,thrarray(0)+"|"+thrarray(0).hashCode,thrarray(2)+"|"+thrarray(3))
    }).cache()
 
    /* (5064062000|2042345,car000044|000044,5|car)
     * 数据处理：对用户产品访问数据进行处理，并生成算法需要的数据格式Rating
	   * 数据样例：用户id(int)，产品id(int)，pv数
	   *  2074472858,000044,1
		    2101313658,000024,6
		    -2132044919,000014,5
	   */
    val ratings = basicId.map(x => {
      //用户id(int)，产品id(int)，pv数
      new Rating(x._1.split("\\|")(1).toInt, x._2.split("\\|")(1).toInt, x._3.split("\\|")(0).toDouble)
    })
 
    /*  算法数据建模，spark固定公式   */
    //矩阵分解的维度，隐含因子的个数越多越好，但越多耗内存更多
    val rank = 5
    //最大迭代次数,默认5，建议值10次左右
    val numIterations = 10
    //推荐模型，用于预测         0.01正则化系数， 此参数正是交叉验证需要验证的参数之一
    val model = ALS.train(ratings, rank, numIterations, 0.01)
 
    //------------------------------------------------------------------------------------------------------
    //user_rdd数据集：表示的是所有用户ID的Hash码的数据    2042345
    val user_rdd = basicId.map(x => x._1.split("\\|")(1).toInt).distinct()
    //pro_rdd数据集： 表示的是所有产品ID的Hash码的数据     000044
    val pro_rdd = basicId.map(x => x._2.split("\\|")(1).toInt).distinct()
 
    //cartesian笛卡尔积     ((2042345,000044))
    val userProScore = user_rdd.cartesian(pro_rdd)
 
    //predict()数据预测,根据模型预测用户对每个产品的评分。int型用户id;int型产品id,评分
    //用户产品对                                        (2042345|00044,0.5)
    val userProPair = model.predict(userProScore).map(x => ("" + x.user + "|" + x.product, x.rating))
 
    //获取用户ID与相对应的Hash码
    // 5064062000|2042345,car000044|000044,5|car
    // car,5064062000|2042345
    val full_user_rdd = basicId.map(x => {
      (x._3.split("\\|")(1), x._1)
    }).distinct
 
    //获取产品ID与相对应的Hash码
    // 5064062000|2042345,car000044|000044,5|car
    //car,car000044|000044
    val full_pro_rdd = basicId.map(x => {
      (x._3.split("\\|")(1), x._2)
    }).distinct
 
    //用户产品join连接
    //car ,(5064062000|2042345,car000044|000044)
    val full_user_pro = full_user_rdd.join(full_pro_rdd).map(x => {
      var saletype = x._1   //car
      var type1 = 1
      saletype match {
        case "car" => type1 = 1
        case "house" => type1 = 2
      }
      //(2042345|00044,5064062000|car00044|1)
      (x._2._1.split("\\|")(1) + "|" + x._2._2.split("\\|")(1),
        x._2._1.split("\\|")(0) + "|" + x._2._2.split("\\|")(0) + "|" + type1)
    })
 
 
    //userProPair=(2042345|00044,0.5)
    //full_user_pro=(2042345|00044,5064062000|car00044|1)
    //userProMsg=2042345|00044,(0.5,5064062000|car00044|1)
    //5064062000|1,car00044|0.5
    val userProMsg = userProPair.join(full_user_pro).map(x => {
      (x._2._2.split("\\|")(0) + "|" + x._2._2.split("\\|")(2),
        x._2._2.split("\\|")(1) + "|" + x._2._1)
    })
 
    /* 数据处理：按评分取前5
   * 样例数据：用户id|预购类型|推荐产品列表
   *  533047060723|2|house000402,house000098,house000403,house000388,house000382
      513047554444|2|house000296,house000117,house000304,house000100,house000402
      683045963832|1|car000300,car000471,car000097,car000382,car000305
      4510341559|1|car000393,car000107,car000105,car000527,car000004
   * */
    //按用户、预购类型分类，每类别取评分前5
    ///userProMsg:  5064062000|1,car00044|0.5
    val userTypeTop = userProMsg.groupByKey.map(x => {
      val ite = x._2.iterator
      val topMap : util.TreeMap[Double, String] = new util.TreeMap[Double, String]
      while (ite.hasNext) {
        val proPirce = ite.next.split("\\|")
        //取正评分,可根据需求调大，值越大喜欢程度越高。
        if (proPirce(1).toDouble > 0) {
          topMap.put(proPirce(1).toDouble, proPirce(0))
        }
        if (topMap.size > 5) {
          topMap.remove(topMap.firstKey)
        }
      }
      //拼接高分产品
      val sb = new StringBuffer
      import scala.collection.JavaConversions._
      for (key <- topMap.keySet) {
        sb.append(topMap.get(key)).append(",")
      }
      x._1 + "|" + sb.toString.substring(0, sb.toString.length - 1)
    })
 
    /** 测试 **/
    userTypeTop.saveAsTextFile("file:///D://测试数据//dx//14.DxMyALS")
    sc.stop()
 
  }
}